/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.nhncorp.neptune.fs;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.NoSuchElementException;
import java.util.Set;
import java.util.TreeSet;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import com.nhncorp.neptune.changelog.ChangeLogClient;
import com.nhncorp.neptune.changelog.ChangeLogInterruptedException;
import com.nhncorp.neptune.changelog.ServerLocationManager;
import com.nhncorp.neptune.changelog.TransactionData;
import com.nhncorp.neptune.changelog.UnmatchedLogException;
import com.nhncorp.neptune.changelog.previous.ChangeLogStatus;
import com.nhncorp.neptune.common.conf.NConfiguration;
import com.nhncorp.neptune.common.exception.ChangeLogException;
import com.nhncorp.neptune.tabletserver.ChangeLog;
import com.nhncorp.neptune.tabletserver.TabletServerIF;
import com.nhncorp.pleiades.client.LockService;

/**
 * Changelog 처리를 위한 파일시스템 클라이언트
 * 현재 구성은 Neptune에서 자체 개발된 고속의 append 가능한 change log용 파일시스템을 별도로 구축하여 
 * 사용하고 있다.
 * @author nhn
 *
 */
public class PipeBasedChangeLogFileSystem implements ChangeLogFileSystemIF {
  static final Log LOG = LogFactory.getLog(PipeBasedChangeLogFileSystem.class);

  private static final int MAX_NUM_TRY = 3;

  static Map<String, ChangeLogFileSystemInfo> fsCache 
    = new HashMap<String, ChangeLogFileSystemInfo>();

  static {
    createPipeCullerThread().start();
  }
  
  static Set<String> staleTabletNameSet = Collections.synchronizedSet(new TreeSet<String>());

  
  // ChangeLogFileSystemInfo의 모든 public 메소드들은
  // 호출하기 전에 쓰레드가 fsCache의 Lock을 획득해야 한다. 
  static class ChangeLogFileSystemInfo {
    LinkedList<ChangeLogClient> clientList = new LinkedList<ChangeLogClient>();
    int numOfUsedPipes = 0;
    private boolean doConsistencyCheck;
    
    ChangeLogFileSystemInfo(boolean consistencyCheck) {
      this.doConsistencyCheck = consistencyCheck;
    }
    
    public void add(ChangeLogClient fs) {
      clientList.addFirst(fs);
      
      if (numOfUsedPipes > 0) {
        numOfUsedPipes--;
      }
    }
    
    public ChangeLogClient removeFirst() {
      ChangeLogClient client = null;
      
      try {
        client = clientList.removeFirst();
      } catch(NoSuchElementException e) {
        return null;
      }

      setConsistencyCheck(client);
      return client;
    }
    
    public void clear() {
      for(ChangeLogClient fs : clientList) {
        try {
          fs.close();
        } catch (IOException e) {
          LOG.debug("exception in closing change log file system", e);
        }
      }
      
      clientList.clear();
      numOfUsedPipes = 0;
    }

    public void setConsistencyCheck(ChangeLogClient client) {
      if (client != null) {
        // 같은  tablet으로 연결된 pipe들이 이미 존재한다면 
        // log file consistency 검사를 수행하지 않는다.
        if (doConsistencyCheck) {
          client.setConsistencyCheck(numOfUsedPipes <= 0);
        }
        numOfUsedPipes++;
      }
    }
  }
  
  static Executor closeExecutor = Executors.newFixedThreadPool(10);

  private static Thread createPipeCullerThread() {
    final int MAX_CLOSE_RETRY_COUNT = 3;
    
    Thread pipeCullerThread = new Thread() {
      public void run() {
        List<ChangeLogClient> removeList = new ArrayList<ChangeLogClient>();

        while (true) {
          try {
            Thread.sleep(5000);

            extractExpiredClient(removeList);

            for (final ChangeLogClient fs : removeList) {
              closeExecutor.execute(new Runnable() {
                public void run() {
                  int numOfTry = 0;
                  while (numOfTry < MAX_CLOSE_RETRY_COUNT) {
                    try {
                      fs.close();
                      break;
                    } catch (IOException e) {
                      if (numOfTry + 1 > MAX_CLOSE_RETRY_COUNT) {
                        LOG.warn("fail to close pipe key[" + fs.getPipeKey() + "]", e);

                        try {
                          fs.disconnect();
                        } catch (Exception ex) {
                          LOG.error("fail to disconnect pipe key [" + fs.getPipeKey() + "]", ex);
                        }

                        break;
                      } else {
                        LOG.warn("fail to close pipe key[" + fs.getPipeKey() + "] due to " + e
                            + ". RETRY");
                      }
                    }

                    try {
                      Thread.sleep(1000);
                    } catch (InterruptedException e) {
                      return;
                    }

                    numOfTry++;
                  }
                }
              });
            }

            removeList.clear();
          } catch (InterruptedException e) {
            LOG.info("Pipe culler thread ends");
            return;
          } catch (Exception e) {
            LOG.error("Exception in pipe culler thread, but continue", e);
          }
        }
      }

      private void extractExpiredClient(List<ChangeLogClient> removeList) {
        synchronized (fsCache) {
          Iterator<Map.Entry<String, ChangeLogFileSystemInfo>> mapIter 
            = fsCache.entrySet().iterator();

          while (mapIter.hasNext()) {
            LinkedList<ChangeLogClient> fsList = mapIter.next().getValue().clientList;

            Iterator<ChangeLogClient> listIter = fsList.iterator();

            while (listIter.hasNext()) {
              ChangeLogClient fs = listIter.next();

              if (fs.isExpired()) {
                if (fs.isClosed() == false) {
                  removeList.add(fs);
                }
                listIter.remove();
              }
            }

            if (fsList.size() == 0) {
              mapIter.remove();
            }
          }
        }
      }
    };

    pipeCullerThread.setDaemon(true);
    return pipeCullerThread;
  }

  NConfiguration conf;
  ServerLocationManager serverLocationManager;
  ThreadLocal<ChangeLogClient> fsThreadLocal = new ThreadLocal<ChangeLogClient>();
  boolean testmode = false;

  // sangchul
  // 이 데이터들은 TabletServer가 처음 기동되면서 changelog를 읽을 때
  // 한 번만 사용되기 때문에 멀티 쓰레드에 의해 접근될 가능성은 없다.
  TransactionData[] dataList;
  int curLogIndex;
  int curDataIndex;
  boolean consistencyCheck = false;

  PipeBasedChangeLogFileSystem(boolean consistencyCheck) {
    this.consistencyCheck = consistencyCheck; 
  }

  public void open(String tabletName, boolean writable) throws IOException {
    if (staleTabletNameSet.contains(tabletName)) {
      LOG.info("tablet [" + tabletName + "] is already in staleTabletNameSet");
      throw new ChangeLogInterruptedException();
    }

    ChangeLogClient fs = fsThreadLocal.get();

    if (fs == null) {
      fs = getFromCache(tabletName);
    }

    if (fs.isClosed() && writable) {
      try {
        fs.open();
        LOG.debug("Pipe for tablet : " + tabletName + " is open");
      } catch(ChangeLogInterruptedException e) {
        throw e;
      } catch(UnmatchedLogException e) {
        registerToStaleTabletNameSet(tabletName);
        throw new ChangeLogException("Unmatched change logs, tabletName [" + tabletName + "]");
      } catch(IOException e) {
        LOG.warn("Fail to open change log filesystem tabletName[" + tabletName + "]", e);
        fs = null;
        registerToStaleTabletNameSet(tabletName);
        throw new ChangeLogException("Unable to open change log filesystem, tabletName ["
            + tabletName + "] due to " + e);
      }

    }

    if (writable) {
      fsThreadLocal.set(fs);
    } else {
      dataList = null;
      curLogIndex = 0;
      curDataIndex = 0;
    
      returnToCache(tabletName, fs);
    }
  }

  private void registerToStaleTabletNameSet(String tabletName) {
    if (staleTabletNameSet.add(tabletName) == false) {
      LOG.info("Tablet [" + tabletName + "] is already in stale set");
    } else {
      LOG.info("put tablet [" + tabletName + "] to stale set");
    }
  }

  // by sangchul
  // fsCache의 monitor lock을 획득한 채로 이 메소드를 호출해야 한다.
  private ChangeLogClient createChangeLogClient(NConfiguration conf2, String tabletName,
      InetSocketAddress[] addresses) throws IOException {
    ChangeLogClient fs = new ChangeLogClient(conf, tabletName, addresses);
    ChangeLogFileSystemInfo info = fsCache.get(tabletName);
    
    if (info == null) {
      info = new ChangeLogFileSystemInfo(consistencyCheck);
      fsCache.put(tabletName, info);
    }
    
    info.setConsistencyCheck(fs);
    
    return fs;
  }

  public void open(String hostName, String tabletName, boolean writable) throws IOException {
    open(tabletName, writable);
  }

  private ChangeLogClient getOpenedChangeLogClient(String tabletName) throws IOException {
    ChangeLogClient fs = null;
    int numTry = 0;

    while((fs = fsThreadLocal.get()) == null || fs.isClosed()) {
      if (++numTry >= MAX_NUM_TRY) {
        throw new IOException("unable to open change log...");
      }

      LOG.info("reopen change log client");
      open(tabletName, true);
    }

    return fs;
  }

  public boolean addChangeLog(String tabletName, String txId, int seq, ChangeLog changeLog)
      throws IOException {
    ChangeLogClient fs = null;

    if ((fs = fsThreadLocal.get()) == null || fs.isClosed()) {
      fs = getOpenedChangeLogClient(tabletName);
      fsThreadLocal.set(fs);
    }
    
    fs.touch();
    if (!fs.isWritingStarted()) {
      try {
        fs.startWriting(txId);
      } catch(ChangeLogInterruptedException e) {
        throw e;
      } catch (IOException e) {
        LOG.warn("fail to start writing changelogs tablet[" + tabletName + "]", e);
        fsThreadLocal.set(null);
        registerToStaleTabletNameSet(tabletName);
        throw new ChangeLogException(e);
      }
    }

    fs.append(changeLog);
    
    return true;
  }

  public void finishAdding(String tabletName, String txId) throws IOException {
    ChangeLogClient fs = null;

    if ((fs = fsThreadLocal.get()) == null || fs.isClosed()) {
      fs = getOpenedChangeLogClient(tabletName);
      fsThreadLocal.set(fs);
    }

    try {
      fs.touch();

      fs.commitWriting();

      returnToCache(tabletName, fs);
    } catch(ChangeLogInterruptedException e) {
      throw e;
    } catch(IOException e) {
      LOG.warn("fail to commit change logs tabletName[" + tabletName + "]", e);
      registerToStaleTabletNameSet(tabletName);
      throw new ChangeLogException(e);
    } finally {
      fsThreadLocal.set(null);
    }
  }

  void returnToCache(String tabletName, ChangeLogClient fs) {
    if (fs == null) {
      Thread.dumpStack();
    }

    synchronized (fsCache) {
      ChangeLogFileSystemInfo fsInfo = fsCache.get(tabletName);
      if (fsInfo == null) {
        fsInfo = new ChangeLogFileSystemInfo(consistencyCheck);
        fsCache.put(tabletName, fsInfo);
      }
      fsInfo.add(fs);
    }
  }

  ChangeLogClient getFromCache(String tabletName) throws IOException {
    ChangeLogClient fs = null;

    synchronized (fsCache) {
      fs = internalRemoveFromCache(tabletName);

      if (fs == null) {
        InetSocketAddress[] addresses = getChangeLogServerInfo(tabletName);
        fs = createChangeLogClient(conf, tabletName, addresses);
      }
    }

    return fs;
  }

  // by sangchul
  // 이 메소드를 호출하는 쓰레드는 fsCache의 모니터를 획득해야 한다.
  private ChangeLogClient internalRemoveFromCache(String tabletName) {
    ChangeLogFileSystemInfo fsInfo = fsCache.get(tabletName);

    if (fsInfo == null) {
      return null;
    }

    ChangeLogClient fs = fsInfo.removeFirst();

    return fs == null ? null : fs.touch();
  }

  public void close(String tabletName, boolean writable) throws IOException {
    synchronized (fsCache) {
      ChangeLogFileSystemInfo fsInfo = fsCache.get(tabletName);

      if (fsInfo != null) {
        fsInfo.clear();
        fsCache.remove(tabletName);
      }
    }
  }

  public void delete(String tabletName) throws IOException {
    ChangeLogClient fs = getFromCache(tabletName);
    try {
      fs.removeAllLogs();
    } finally {
      returnToCache(tabletName, fs);
    }
  }

  public void startMinorCompaction(String tabletName) throws IOException {
    ChangeLogClient fs = getFromCache(tabletName);
    try {
      fs.backupLogFile();
    } finally {
      returnToCache(tabletName, fs);
    }
  }

  public void endMinorCompaction(String tabletName) throws IOException {
    if (staleTabletNameSet.remove(tabletName)) {
      removeChangeLogServerInfo(tabletName);
    }

    ChangeLogClient fs = getFromCache(tabletName);
    
    try {
      //fs.removeBackupLogFiles();
    } finally {
      returnToCache(tabletName, fs);
    }
  }

  public boolean exists(String tabletName) throws IOException {
    ChangeLogClient fs = getFromCache(tabletName);
    try {
      return fs.tabletExists() ? fs.getLogFileSize() > 0 : false;
    } catch(UnmatchedLogException e) {
      LOG.warn("Existence check of tablet[" + tabletName + "] returns TRUE even though exception occurs :" + e);
      // by sangchul
      // if the size of log files are different, existing check should return true value 
      return true;    
    } finally {
      returnToCache(tabletName, fs);
    }
  }

  public void format() throws IOException {
    for (InetSocketAddress addr : serverLocationManager.getAllAddresses()) {
      ChangeLogClient.formatLogsAt(conf, addr);
    }
  }

  public InetSocketAddress[] getChangeLogServerInfo(String tabletName) throws IOException {
    return serverLocationManager.getReplicaAddresses(tabletName);
  }

  private void removeChangeLogServerInfo(String tabletName) throws IOException {
    serverLocationManager.removeReplica(tabletName);
  }

  public void init(ChangeLogFileSystemIF systemIF, NConfiguration conf,
      TabletServerIF tabletServer, LockService lockService) throws IOException {
    this.conf = conf;
    this.testmode = Boolean.valueOf(conf.get("testmode", "false"));
    this.serverLocationManager = ServerLocationManager.instance(conf);
  }

  public ChangeLog nextChangeLog(String tabletName) throws IOException {
    if (dataList == null) {
      loadChangeLog(tabletName);

      if (dataList == null || dataList.length == 0) {
        return null;
      }
    }

    if (curLogIndex >= dataList[curDataIndex].getChangeLogLength()) {
      curLogIndex = 0;
      curDataIndex++;
    }

    if (curDataIndex >= dataList.length) {
      return null;
    }

    ChangeLog ret = dataList[curDataIndex].getChangeLogAt(curLogIndex);
    curLogIndex++;

    return ret;
  }

  private void loadChangeLog(String tabletName) throws IOException {
    ChangeLogClient fs = getFromCache(tabletName);

    try {
      dataList = fs.readLastLogFile();
    } finally {
      returnToCache(tabletName, fs);
    }
    curLogIndex = 0;
    curDataIndex = 0;
  }

  public ChangeLogStatus verifyChangelog(String tabletName) throws IOException {
    ChangeLogStatus status = new ChangeLogStatus();
    long size = 0;
    ChangeLogClient fs = getFromCache(tabletName);
    boolean isInconsistent = false;
    try {
      isInconsistent = fs.hasInvalidChangeLog();
      status.setNeedCompaction(isInconsistent);
      
      if (isInconsistent) {
        LOG.warn("ChangeLogs of tablet [" + tabletName + "] is inconsistent");
      }
    } catch (Exception e) {
      LOG.warn("Exception in verifying change logs", e);
      status.setNeedCompaction(true);
    } finally {
      returnToCache(tabletName, fs);
    }

    loadChangeLog(tabletName);

    if (dataList == null) {
      LOG.warn("Strange in loading change logs of tablet[" 
          + tabletName + "]. Return empty logs");
      return new ChangeLogStatus(); // return empty status object
    }

    if (dataList.length > 0) {
      TransactionData lastData = dataList[dataList.length - 1];
      status.setMaxSeq(lastData.getSeqNo());
      status.setMaxTxId(new String(lastData.getID()).trim());
    }
    status.setFileSize(size);
    return status;
  }

  public String getTestHandlerKey() {
    LOG.fatal("NOT IMPLEMENTED");
    return null;
  }

  public List<ChangeLogStatus> getChangeLogStatus(String tabletName) throws IOException {
    // TODO Auto-generated method stub
    return null;
  }
}
